package com.hatricks.is.base.stream.progressinputstream;

import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;

import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;

// 进度输入流
/**
 * @author :yanyh
 * @date: 2024年3月27日14:29:41
 */
@Slf4j
public class ProgressInputStream extends FilterInputStream {
    /**
     * 默认通知阈值8KB
     */
    private static final int DEFAULT_NOTIFICATION_THRESHOLD = 8 * 1024;
    private final ProgressListener listener;
    private final int notifyThresHold;
    private int unModifiedByteCount;
    private boolean hasBeenRead;
    private boolean doneEOF;
    private long notifiedByteCount;

    public ProgressInputStream(InputStream is, ProgressListener listener) {
        this(is, listener, DEFAULT_NOTIFICATION_THRESHOLD);
    }

    public ProgressInputStream(InputStream is, ProgressListener listener, int notifyThresHold) {
        super(is);
        if (is == null || listener == null) {
            throw new IllegalArgumentException();
        }
        this.listener = listener;
        this.notifyThresHold = notifyThresHold;
    }

    protected void onFirstRead() {
        ProgressPublisher.publishByteCountEvent(listener, ProgressEventType.TRANSFER_START, 1);
    }

    protected void onEOF() {
        ProgressPublisher.publishByteCountEvent(listener, ProgressEventType.TRANSFER_FINISHED, getNotifiedByteCount() + getUnModifiedByteCount());
    }

    protected void onClose() {
        // 会调用到EOF
        eof();
    }

    protected void onReset() {
        log.warn("暂时不通知rest类型事件");
    }

    protected void onNotifyBytesRead() {
        ProgressPublisher.publishByteCountEvent(listener, ProgressEventType.TRANSFER_CHANGED, getNotifiedByteCount());
    }

    private void onBytesRead(int bytesRead) {
        unModifiedByteCount += bytesRead;
        if (unModifiedByteCount >= notifyThresHold) {
            notifiedByteCount += unModifiedByteCount;
            unModifiedByteCount = 0;
            onNotifyBytesRead();
        }
    }

    @Override
    public int read() throws IOException {
        if (!hasBeenRead) {
            onFirstRead();
            hasBeenRead = true;
        }
        int ch = super.read();
        if (ch == -1)
            eof();
        else
            onBytesRead(1);
        return ch;
    }

    @Override
    public void reset() throws IOException {
        super.reset();
        onReset();
        unModifiedByteCount = 0;
        notifiedByteCount = 0;
    }

    @Override
    public int read(@NotNull byte[] b) throws IOException {
        return read(b, 0, b.length);
    }

    @Override
    public int read(@NotNull byte[] b, int off, int len) throws IOException {
        if (!hasBeenRead) {
            onFirstRead();
            hasBeenRead = true;
        }
        int bytesRead = super.read(b, off, len);
        if (bytesRead == -1)
            eof();
        else
            onBytesRead(bytesRead);
        return bytesRead;
    }

    /**
     * 流数据读完或者close方法调用即位结束
     */
    private void eof() {
        if (doneEOF)
            return;
        onEOF();
        unModifiedByteCount = 0;
        doneEOF = true;
    }

    public final InputStream getWrappedInputStream() {
        return in;
    }

    protected final int getUnModifiedByteCount() {
        return unModifiedByteCount;
    }

    protected final long getNotifiedByteCount() {
        return notifiedByteCount;
    }

    @Override
    public void close() throws IOException {
        onClose();
        super.close();
    }

    public final ProgressListener getListener() {
        return listener;
    }
}
